package com.kingsoft.dc.khaos.listener

import java.util.concurrent.{LinkedBlockingQueue, Semaphore}
import java.util.concurrent.atomic.AtomicBoolean

/**
  * Created by chengguangqing on 2019/5/28.
  */
private[khaos] class LiveListenerBus extends KhaosListenerBus{

  private val EVENT_QUEUE_CAPACITY = 10000

  private val eventQueue =
    new LinkedBlockingQueue[KhaosListenerEvent](EVENT_QUEUE_CAPACITY)

  private val started = new AtomicBoolean(false)
  private val stopped = new AtomicBoolean(false)

  private var processingEvent = false
  private val eventLock = new Semaphore(0)
  private val logDroppedEvent = new AtomicBoolean(false)

  private val listenerThread = new Thread(LiveListenerBus.name){
    setDaemon(true)

    override def run(): Unit = {
      while(true){
        eventLock.acquire()
        this.synchronized{
          processingEvent = true
        }

        try{
          val event = eventQueue.poll()
          if(event == null){
            if(!stopped.get){
              throw new IllegalStateException(
                s"""
                   |Polling `null` from eventQueue means
                   |the listener bus has been stopped
                   |So `stopped` must be true
                 """.stripMargin
              )
            }
            return
          }
          postToAll(event)
        } finally{
          this.synchronized{
            processingEvent = false
          }
        }
      }
    }
  }
}

private[khaos] object LiveListenerBus {
  val name = "Khaos-Listener-Bus"
}
